package com.zhkj.crawl.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.zhkj.crawl.bean.CrawAmzParams;
import com.zhkj.crawl.bean.FetchPhpParams;
import com.zhkj.crawl.config.Config;
import com.zhkj.crawl.config.Constants;
import com.zhkj.crawl.config.FetchRankException;
import com.zhkj.crawl.config.StringPool;
import com.zhkj.crawl.entity.Listing;
import com.zhkj.crawl.entity.ResultAmz;
import com.zhkj.crawl.rank.AmzService;
import com.zhkj.crawl.rank.BaseService;
import com.zhkj.crawl.service.TransactionService;
import com.zhkj.crawl.utils.AmzRankStrFinder;
import com.zhkj.crawl.utils.TimeDistance;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.jsoup.nodes.Element;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.stereotype.Service;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

/**
 * 业务方法
 */
@Service
@Slf4j
public class BusService {

    @Autowired
    private ListingService listingService;

    @Autowired
    private RetryService retryService;

    @Autowired
    private ResultAmzService resultAmzService;

    @Autowired
    private PhpService phpService;

    @Autowired
    private TransactionService transactionService;
    private static Integer MAX_THREAD = 3;
    private static Integer MAX_CORE_THREAD = MAX_THREAD;
    public static final ThreadPoolExecutor exec = new ThreadPoolExecutor(MAX_CORE_THREAD, MAX_THREAD,
            60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(4000), new CustomizableThreadFactory("crawlThread-"));

    public static void modifyThreadSize(int size){
        exec.setCorePoolSize(size);
        exec.setMaximumPoolSize(size);

    }

    /**
     * 入口
     * 1.尝试抓取listing
     * 1.1.尝试待重试listing
     * 3.尝试结果表重试
     */
    public void enter(CrawAmzParams params){
        Config.TB_CURRENT = params.getTb();
        Config.SITE_CURRENT = params.getSite();

        Config.rate = new SuccessRate();
        List<Listing> result = listingService.getUndoList(Config.limit, params.getPlat());
        if(CollUtil.isNotEmpty(result)){
            doCrawler(result);
            return;
        }

        List<ResultAmz> results = resultAmzService.getRetryList(Config.limit);
        if(CollUtil.isNotEmpty(results)){
            retryService.doRetry(results);
            return;
        }
    }

    public void doPushPhp(){
        Config.TB_CURRENT = "_php";
        List<ResultAmz> doneList = resultAmzService.getDoneList();
        if(CollUtil.isNotEmpty(doneList)){
            phpService.pushResult(doneList);
            resultAmzService.updatePushState();
        }
    }

    public void doCrawler(List<Listing> result){
        log.info("开始抓取listing {}条", result.size());
        int total = result.size();
        Config.rate.setTotal(total);
        List<List<Listing>> partition = split(result);

        // 假设单条最大等待40s
        long timeout = partition.size() * (partition.get(0).size() * 40L);
        TimeDistance td = new TimeDistance();
        try {
            CompletableFuture.allOf(
                    partition.stream()
                            .map(s -> CompletableFuture.supplyAsync(() -> handlerList(s), exec))
                            .toArray(CompletableFuture[]::new)
            ).get(timeout, TimeUnit.SECONDS);
        } catch (Exception e) {
            log.error("执行错误", e);
        }
        td.dot();
        log.info("总条数{} 执行耗时{} 代理{} 结果{}", total, td.pop(), Config.proxySwitch, Config.rate);
    }

    public static <T> List<List<T>> split(List<T> result){
        // 条数1000 / 线程数20 = 50批次
        // 批次最大500，超过则进入线程队列等待
        int total = result.size();
        int batchSize = total < MAX_THREAD ? total :
                total % MAX_THREAD > 0 ? total / MAX_THREAD + 1 : total / MAX_THREAD;
        batchSize = Math.min(batchSize, Config.MAX_BATCH);

        List<List<T>> partition = Lists.partition(result, batchSize);
        int partitionSize = partition.size();
        log.info("总条数{}=单批次数{} * 批次{} 代理{}", total, batchSize, partitionSize, Config.proxySwitch);
        return partition;
    }

    public void fetchData(FetchPhpParams params){
        Config.TB_CURRENT = "_php";
        Config.scheduling = true;
        String mon = Constants.getCurMon();
        log.info("{}开始同步", mon);
        int i = phpService.syncPhpListing(params);
        log.info("{}本次同步{}条数据", mon, i);
        Config.scheduling = false;
    }

    private String handlerList(List<Listing> part) {
        Map<String, List<Listing>> group = part.stream().collect(Collectors.groupingBy(Listing::getPlat));

        for (Map.Entry<String, List<Listing>> entry : group.entrySet()) {
            switch (entry.getKey()) {
                case "亚马逊":
                    handlerAmz(entry.getValue());
                    break;
                case "速卖通":
                case "京东国际":
                case "WALMART":
                default:
                    break;
            }
        }
        return null;
    }


    public void handlerAmz(List<Listing> value) {
        log.info("开始处理{}条亚马逊数据", value.size());

        int size = value.size(), q = 0;
        List<ResultAmz> results = new ArrayList<>();
        String month = Constants.getCurMon();
        List<Long> errIds = new ArrayList<>();

        try {
            for (Listing listing : value) {
                ++q;
                listing.setLastOptMonth(month);
                if(StringUtils.isAnyBlank(listing.getPlatId(), listing.getSite())){
                    listing.setStatus(Listing.Status.DATA_ERR.ordinal());
                    listing.setRemark("无效数据");
                    continue;
                }
                singleAmz(errIds, listing, q, size, month, results);
            }
        }finally {
            ChromeService.release();
        }

        try {
            Config.rate.addSuccess(results.size());
            Config.rate.addHasRank((int) results.stream().filter(s -> StringUtils.isNotBlank(s.getRankStr())).count());

            List<Listing> updateListing = value.stream().filter(s -> !errIds.contains(s.getId())).collect(Collectors.toList());
            transactionService.updateResultAmz(updateListing, results);
        }catch (Exception e){
            List<String> batchRes = results.stream().map(s -> String.format("%s_%s", s.getSite(), s.getAsin())).collect(Collectors.toList());
            log.error(String.format("更新结果出错 %s", batchRes), e);
        }
    }


    private void singleAmz(List<Long> errIds, Listing listing, int q, int size, String month, List<ResultAmz> results) {
        try {
            BaseService.ProductDetail detail = AmzService.start(listing.getPlatId(), listing.getSite());

            Element element = detail.getElement();
            detail.setElement(null);
            log.info("{} 的详情({}/{})：{}", detail.getItemId(), q, size, JSON.toJSONString(detail));
            // 啥都没爬到 则改状态下次重试
            if(StringUtils.isAllBlank(detail.getRank(), detail.getRating(), detail.getStar())){
                listing.setStatus(Listing.Status.RETRY.ordinal());
                int alreadyTry = 0;
                try { alreadyTry = Integer.parseInt(listing.getRemark()); }catch (Exception ignore){}
                listing.setRemark((alreadyTry + 1) + "");
                return;
            }

            ResultAmz result = new ResultAmz();
            result.setAsin(listing.getPlatId());
            result.setSite(listing.getSite());
            result.setRankStr(detail.getRank());
            result.setStarStr(detail.getStar());
            result.setRatingStr(detail.getRating());
            result.setMonth(month);
            result.setUrl(detail.getUrl());
            result.setListingId(listing.getId());
            result.setTimes(1);
            AmzRankStrFinder.findAndFillNum(result);
            results.add(result);

            // 爬到了 则改状态为初始值
            listing.setStatus(Listing.Status.INIT.ordinal());
            listing.setRemark(StringPool.EMPTY);
        }catch (FetchRankException e1){
            FetchRankException.Type errType = e1.getErrType();
            if(errType.equals(FetchRankException.Type.INVALID)){
                listing.setStatus(Listing.Status.RES_INVALID.ordinal());
                listing.setRemark(errType.msg);
            }else if(errType.equals(FetchRankException.Type.NONE_INFO)){
                listing.setStatus(Listing.Status.NONE_INFO.ordinal());
                listing.setRemark(errType.msg);
            }else {
                errIds.add(listing.getId());
                log.error(String.format("amzStartErr1(%s)", listing.getPlatId()), e1);
            }
            log.info("{} 无效 {} ({})", listing.getPlatId(), e1.getErrType(), e1.getUrl());
        }catch (Exception e){
            errIds.add(listing.getId());
            log.error(String.format("amzStartErr2(%s)", listing.getPlatId()), e);
        }
    }

    @Data
    public static class SuccessRate{
        private int total;
        private int success;
        private int hasRank;

        public SuccessRate(int total, int success, int hasRank) {
            this.total = total;
            this.success = success;
            this.hasRank = hasRank;
        }

        public SuccessRate() { }

        private String getSuccessRate(){
            return new BigDecimal(success)
                    .divide(new BigDecimal(total), 2, RoundingMode.HALF_UP)
                    .multiply(BigDecimal.valueOf(100)) + "%";
        }

        private String getRankRate(){
            return new BigDecimal(hasRank)
                    .divide(new BigDecimal(total), 2, RoundingMode.HALF_UP)
                    .multiply(BigDecimal.valueOf(100)) + "%";
        }

        @Override
        public String toString() {
            final StringBuffer sb = new StringBuffer("{");
            sb.append("total=").append(total);
            sb.append(", success=").append(success).append(" ").append(getSuccessRate());
            sb.append(", hasRank=").append(hasRank).append(" ").append(getRankRate());
            sb.append('}');
            return sb.toString();
        }

        public synchronized void addSuccess(int success) {
            this.success += success;
        }

        public synchronized void addHasRank(int hasRank) {
            this.hasRank += hasRank;
        }
    }

}
